package com.study.chapter11;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/25 21:14
 */
public class UdfTableAggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

        //1.在创建表的DDL中直接定义时间属性：事件时间
        String createDDL = " CREATE TABLE click_table (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " event_time AS TO_TIMESTAMP( FROM_UNIXTIME( ts / 1000 ) ), " +   //将ts时间戳转换成timestamp类型
                " WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND " +  //设置水位线的时间间隔为1s
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";
        tabEnv.executeSql(createDDL);

        //2.注册自定义表聚合函数
        tabEnv.createTemporarySystemFunction("Top2Function",Top2Function.class);

        //3.调用UDF进行查询转换
        String windAggQuery = " select user,count(url) as cnt,window_start,window_end " +
                " from table(" +
                " TUMBLE( TABLE click_table, DESCRIPTOR(event_time),INTERVAL '10' SECOND )  " +  //滚动窗口
                " ) " +
                " group by user,window_start,window_end";
        Table aggTable = tabEnv.sqlQuery(windAggQuery);

        Table resultTable = aggTable.groupBy($("window_end"))
                .flatAggregate(
                        call("Top2Function", $("cnt"))
                                .as("value", "rank")
                )
                .select($("window_end"), $("value"), $("rank"));

        //4.转换成流打印输出
        tabEnv.toChangelogStream(resultTable).print("result：");

        env.execute();
    }

    //单独定义一个累加器类型，包含了当前最大和第二大的数据
    public static class Top2Accumulator{
        public Long max;
        public Long secondMax;
    }

    //实现一个自定义的表聚合函数
    public static class Top2Function extends TableAggregateFunction<Tuple2<Long,Integer>, Top2Accumulator>{

        @Override
        public Top2Accumulator createAccumulator() {
            Top2Accumulator accumulator = new Top2Accumulator();
            accumulator.max = Long.MIN_VALUE;
            accumulator.secondMax = Long.MIN_VALUE;
            return accumulator;
        }

        //定义一个更新累加器的方法
        public void accumulate(Top2Accumulator accumulator,Long value){
            if (value > accumulator.max){
                accumulator.secondMax = accumulator.max;
                accumulator.max = value;
            }else if (value > accumulator.secondMax){
                accumulator.secondMax = value;
            }
        }

        //输出结果，当前的top2
        public void emitValue(Top2Accumulator accumulator, Collector<Tuple2<Long,Integer>> out){
            if (accumulator.max != Long.MIN_VALUE){
                out.collect(Tuple2.of(accumulator.max,1));
            }
            if (accumulator.secondMax != Long.MIN_VALUE){
                out.collect(Tuple2.of(accumulator.secondMax,2));
            }
        }
    }
}
